package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant.*;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.JDBCUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.HashMap;

import static com.atguigu.realtime.common.Constant.*;

/**
 * @author chenlongStart
 * @create 2021-06-28 14:55
 * @descERROR Exception occurred in REST handler: Job c79153e78fa6d637b368aabb8da61503 not found
 */
public class DwmOrderWideApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideApp().init(3003, 1, "DwmOrderWideApp", TOPIC_DWD_ORDER_INFO, "DwmOrderWideApp", TOPIC_DWD_ORDER_DETAIL);
    }

    @Override
    public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> sourceStream) {
        // 1. 两个事实表进行join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = joinFact(sourceStream);

        // 2. join维度信息
        joinDim(orderWideStreamWithoutDims);

    }

    private void joinDim(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
        orderWideStreamWithoutDims
                .map(new RichMapFunction<OrderWide, OrderWide>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JDBCUtil.getJdbcConnection(PHOENIX_DRIVER, PHOENIX_URL);
                    }

                    @Override
                    public OrderWide map(OrderWide orderWide) throws Exception {
                        //dim_user_info
                        JSONObject userInfo = DimUtil.readDimFromPhoenix(conn, DIM_USER_INFO, orderWide.getUser_id());
                        orderWide.setUser_gender(userInfo.getString("GENDER"));
                        orderWide.calcUserAgeByBirthday(userInfo.getString("BIRTHDAY"));
                        orderWide.setOperate_time(userInfo.getString("OPERATE_TIME"));

                        //dim_base_province
                        JSONObject provinceInfo = DimUtil.readDimFromPhoenix(conn,DIM_BASE_PROVINCE, orderWide.getProvince_id());
                        orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
                        orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
                        orderWide.setProvince_name(provinceInfo.getString("NAME"));
                        orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));

                        //dim_sku_info
                        JSONObject skuInfo = DimUtil.readDimFromPhoenix(conn,DIM_SKU_INFO, orderWide.getSku_id());
                        orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
                        orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));
                        orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
                        orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                        orderWide.setTm_id(skuInfo.getLong("TM_ID"));

                        //dim_spu_info
                        JSONObject spuInfo = DimUtil.readDimFromPhoenix(conn,DIM_SPU_INFO, orderWide.getSpu_id());
                        orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));

                        //dim_base_category3
                        JSONObject category3Info = DimUtil.readDimFromPhoenix(conn,DIM_BASE_CATEGORY3, orderWide.getCategory3_id());
                        orderWide.setCategory3_name(category3Info.getString("NAME"));

                        //dim_base_trademark
                        JSONObject trademarkInfo = DimUtil.readDimFromPhoenix(conn,DIM_BASE_TRADEMARK, orderWide.getTm_id());
                        orderWide.setTm_name(trademarkInfo.getString("TM_NAME"));

                        return orderWide;
                    }

                    @Override
                    public void close() throws Exception {
                        if (conn != null) {
                            conn.close();
                        }
                    }
                })
                .print();
    }

    private SingleOutputStreamOperator<OrderWide> joinFact(HashMap<String, DataStreamSource<String>> sourceStream) {
        KeyedStream<OrderInfo, Long> orderInfoStream = sourceStream
                .get(TOPIC_DWD_ORDER_INFO)
                .map(json -> JSON.parseObject(json, OrderInfo.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((orderInfo, ts) -> orderInfo.getCreate_ts())
                )
                .keyBy(OrderInfo::getId);

        KeyedStream<OrderDetail, Long> orderDetailStream = sourceStream
                .get(TOPIC_DWD_ORDER_DETAIL)
                .map(json -> JSON.parseObject(json, OrderDetail.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((orderDetail, ts) -> orderDetail.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);

        return orderInfoStream
                .intervalJoin(orderDetailStream)
                .between(Time.minutes(-5), Time.minutes(5))
                // join 完成应该是返回一张宽表, 这张宽表目前维度只有一些id , 其实是缺少一些维度信息
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo left,
                                               OrderDetail right,
                                               Context ctx,
                                               Collector<OrderWide> out) throws Exception {
                        out.collect(new OrderWide(left, right));
                    }
                });
    }
}
